from kafka import TopicPartition
import json
import jsonpath
if __name__ == '__main__':
    # CN_BOOT_SERVERS = '10.19.255.238:9092,10.19.255.138:9092,10.19.255.142:9092'
    # CN_BOOT_SERVERS = '10.19.255.181:9092,10.19.255.188:9092,10.19.255.189:9092'
    CN_BOOT_SERVERS = '10.100.64.38:9092'
    # CN_BOOT_SERVERS = 'wkafka11.wisers.com:9092,wkafka12.wisers.com:9092,wkafka13.wisers.com:9092'
    # US_BOOT_SERVERS = "10.3.1.124:9092,10.3.1.185:9092,10.3.1.32:9092"

    BOOTSTRAP_SERVERS = CN_BOOT_SERVERS.split(',')
    PARTITION = 1
    consumer = KafkaConsumer(group_id="testttt",
                             bootstrap_servers=BOOTSTRAP_SERVERS)
    # partition = TopicPartition('app.crawler.listing.output', partition=PARTITION)
    partition = TopicPartition('app.crawler.listing.output', partition=PARTITION)
    # partition = TopicPartition('demo.incoming', partition=PARTITION)
    OFFSET = 132733474
    consumer.assign([partition])
    consumer.seek(partition=partition, offset=OFFSET)
    # res = consumer.poll(0)122223514
    # consumer.seek_to_end()
    for msg in consumer:
        # msg = msg.encode()
        message = msg.value
        message = message.decode("utf-8")
        message = json.loads(message)
        print(message)